package rpc

import (
	"context"
	"errors"
	"net"
	"reflect"
	"strconv"
	"time"

	"gitee.com/youkelike/rpc/message"
	"gitee.com/youkelike/rpc/message/compress"
	"gitee.com/youkelike/rpc/message/serialize"
	"gitee.com/youkelike/rpc/message/serialize/json"
	"github.com/silenceper/pool"
)

// service 的实现是一个 struct
// 用代理模式，将 service 的方法类型字段值替换成 rpc 调用
func setFuncField(service Service, p Proxy, s serialize.Serializer, c compress.Compressor) error {
	if service == nil {
		return errors.New("rpc: 不支持 nil")
	}

	val := reflect.ValueOf(service)
	typ := val.Type()
	// 这样用也可以
	// typ := reflect.TypeOf(service)
	if typ.Kind() != reflect.Pointer || typ.Elem().Kind() != reflect.Struct {
		return errors.New("rpc: 只支持指向结构体的一级指针")
	}
	// 拿到指针指向的结构体
	val = val.Elem()
	typ = typ.Elem()

	numField := typ.NumField()
	// numField := val.NumField()
	for i := 0; i < numField; i++ {
		fieldTyp := typ.Field(i)
		fieldVal := val.Field(i)

		if !fieldVal.CanSet() {
			continue
		}
		// 代理模式下提供给函数类型字段的新值
		fn := func(args []reflect.Value) (results []reflect.Value) {
			retVal := reflect.New(fieldTyp.Type.Out(0).Elem())

			// 把参数从 reflect.Value 类型转换成 any 类型
			// 用 any 类型主要是为了方便编码传输
			// 构造参数
			ctx := args[0].Interface().(context.Context)
			// 序列化
			arg, err := s.Encode(args[1].Interface())
			if err != nil {
				return []reflect.Value{retVal, reflect.ValueOf(err)}
			}
			// 压缩
			arg, err = c.Compress(arg)
			if err != nil {
				return []reflect.Value{retVal, reflect.ValueOf(err)}
			}

			meta := make(map[string]string, 2)
			if isOneway(ctx) {
				meta["one-way"] = "true"
			}
			if deadline, ok := ctx.Deadline(); ok {
				meta["deadline"] = strconv.FormatInt(deadline.UnixMilli(), 10)
			}

			req := &message.Request{
				ServiceName: service.Name(),
				MethodName:  fieldTyp.Name,
				Data:        arg,
				Serializer:  s.Code(),
				Compresser:  c.Code(),
				Meta:        meta,
			}

			// rpc 调用服务端方法
			resp, err := p.Invoke(ctx, req)
			if err != nil {
				return []reflect.Value{retVal, reflect.ValueOf(err)}
			}

			var serviceErr error
			if len(resp.Err) > 0 {
				// 业务错误
				serviceErr = errors.New(string(resp.Err))
			}

			if len(resp.Data) > 0 {
				resp.Data, err = c.Decompress(resp.Data)
				if err != nil {
					// 反序列化错误
					return []reflect.Value{retVal, reflect.ValueOf(err)}
				}

				err = s.Decode(resp.Data, retVal.Interface())
				if err != nil {
					// 反序列化错误
					return []reflect.Value{retVal, reflect.ValueOf(err)}
				}
			}

			var serviceErrVal reflect.Value
			if serviceErr == nil {
				serviceErrVal = reflect.Zero(reflect.TypeOf(new(error)).Elem())
			} else {
				serviceErrVal = reflect.ValueOf(serviceErr)
			}

			return []reflect.Value{retVal, serviceErrVal}
		}

		// 将赋给函数字段的新值转换成指定类型
		fnVal := reflect.MakeFunc(fieldTyp.Type, fn)
		// 替换字段值
		fieldVal.Set(fnVal)
	}

	return nil
}

const numOfLengthBytes = 8

type Client struct {
	pool       pool.Pool
	serializer serialize.Serializer
	compress   compress.Compressor
}

type ClientOption func(c *Client)

func ClientWithSerializer(sl serialize.Serializer) ClientOption {
	return func(c *Client) {
		c.serializer = sl
	}
}

func ClientWithCompressor(compress compress.Compressor) ClientOption {
	return func(c *Client) {
		c.compress = compress
	}
}

func NewClient(addr string, ops ...ClientOption) (*Client, error) {
	p, err := pool.NewChannelPool(&pool.Config{
		InitialCap:  1,
		MaxCap:      30,
		MaxIdle:     10,
		IdleTimeout: time.Minute,
		Factory: func() (interface{}, error) {
			return net.DialTimeout("tcp", addr, time.Second*3)
		},
		Close: func(i interface{}) error {
			return i.(net.Conn).Close()
		},
	})
	if err != nil {
		return nil, err
	}
	res := &Client{
		pool:       p,
		serializer: &json.Serializer{},
		compress:   compress.DoNothingCompressor{},
	}
	for _, op := range ops {
		op(res)
	}
	return res, nil
}

func (c *Client) RegisterService(service Service) error {
	return setFuncField(service, c, c.serializer, c.compress)
}

// 只是在 doInvoke 方法外围加了一层超时控制逻辑，
// 这里后面的逻辑只有数据的收发和编解码
func (c *Client) Invoke(ctx context.Context, req *message.Request) (*message.Response, error) {
	var resp *message.Response
	var err error
	ch := make(chan struct{})
	// 为了控制超时才写成这样，doInvoke 方法虽然有 ctx 参数，但其实并没有用它进行超时判断
	go func() {
		resp, err = c.doInvoke(ctx, req)
		select {
		case ch <- struct{}{}:
		default:
		}
	}()
	select {
	case <-ctx.Done():
		return nil, ctx.Err()
	case <-ch:
		return resp, err
	}
}

// 处理数据的编码
func (c *Client) doInvoke(ctx context.Context, req *message.Request) (*message.Response, error) {
	data := message.EncodeReq(req)

	resp, err := c.Send(ctx, data)
	if err != nil {
		return nil, err
	}
	// rpc 方法执行成功正常返回，第二个参数为 nil
	return message.DecodeResp(resp), nil
}

// 处理数据的收发
func (c *Client) Send(ctx context.Context, data []byte) ([]byte, error) {
	val, err := c.pool.Get()
	if err != nil {
		return nil, err
	}
	conn := val.(net.Conn)
	defer func() {
		_ = c.pool.Put(conn)
	}()

	_, err = conn.Write(data)
	if err != nil {
		return nil, err
	}

	if isOneway(ctx) {
		return nil, errors.New("micro: oneway 调用")
	}

	return ReadMsg(conn)
}
